Amazon Cognitoストリームを試してみた

suzuki.ryo

2015.03.06

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

AWSチームのすずきです。

Amazon Cognitoではユーザー認証機能だけでなく、ユーザデータの保管が可能な同期ストアが提供されます。

先日リリースされたAmazon Cognitoストリームを利用する事で、Cognito同期ストアの更新情報をKinesisに流し込む事が可能になりました。

今回、Amazon Cognitoストリームを有効化し、Kinesisに登録されたレコードを確認する事ができましたので、紹介させていただきます。

【AWS発表】Amazon Cognitoストリーム

Cognito設定

Cognito Identity Poolの作成

2015年現在、Cognito利用可能なUS-EAST（N.Virsinia）リージョンでIdentity Poolの作成を行います。

今回はテストの為「Unauth」、認証プロバイダとの連携しないCognito利用を可能とします。

IAMロールは新規作成します

Identity Poolの設定変更画面を開きます。

Kinesisストリームの作成と、Kinesis更新用のIAM設定を行います。

Kinesisのストリーム名とシャード数を指定します。

Kinesis更新用のIAM設定を行います。

「Allow」により、作成したKinesisストリームへのPUT権限が付与されます。

ストリームを有効化し、反映します。

IAM、Kinesisの確認

Identity Poolの作成により追加された、IAMロール、Kinesisストリームを確認します。

Cognito IdentityIdの取得

AWS CLIを利用し、Cognito IdentityIdの取得を取得します。
cognito権限はEC2ロールで付与済とします。

$ AWS_DEFAULT_REGION=`curl -s http://169.254.169.254/latest/meta-data/placement/availability-zone | sed -e 's/.$//g'`
$ export AWS_DEFAULT_REGION

$  aws cognito-identity list-identity-pools --max-results 10
{
    "IdentityPools": [
        {
            "IdentityPoolId": "us-east-1:aec7eb3d-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
            "IdentityPoolName": "test0306"
        }
    ]
}

$ aws cognito-identity get-id  --account-id 784693731708 --identity-pool-id us-east-1:aec7eb3d-xxxx-xxxx-xxxx-xxxxxxxxxxxx
{
    "IdentityId": "us-east-1:27189d5e-yyyy-yyyy-yyyy-yyyyyyyyyyyyy"
}

Cognito 同期ストア操作

テスト用のDatasetを作成します。

Datasetにレコードを追加します。

レコードの更新を行います。

レコードの削除を行います。

レコードの更新、削除後は同期操作が必要です。

レコード、JSON形式で登録する事も可能です。

Kinesis操作

Kinesis 接続確認

AWS CLIを利用して、KinesisのAPIが実行可能な事を確認します。
KinesisのAPI権限はEC2ロールで付与済とします。

$ AWS_DEFAULT_REGION=`curl -s http://169.254.169.254/latest/meta-data/placement/availability-zone | sed -e 's/.$//g'`
$ export AWS_DEFAULT_REGION

$ aws kinesis list-streams 
{
    "StreamNames": [
        "stream0306"
    ]
}

iterator取得

get-shard-iteratorを実行します。
設置直後のKinesisストリームが利用されるものとし、shard-iterator-typeは「TRIM_HORIZON」で取得します。

$ KINESIS_STREAM="stream0306"
$ KINESIS_SHARDID="shardId-000000000000"
$ KINESIS_SHARD_ITERATOR=`aws kinesis get-shard-iterator \
      --stream-name "${KINESIS_STREAM}" \
      --shard-id "${KINESIS_SHARDID}" \
      --shard-iterator-type TRIM_HORIZON \
     | jq -r '.ShardIterator'`

records取得

get-recordsを実行します。
base64でエンコードされたData、デコードします。
取得したshard-iteratorは5分で失効するため、get-recordsの結果は一旦変数に反映します。

$ aws kinesis get-records --shard-iterator "${KINESIS_SHARD_ITERATOR}"  
{
  "Records": [
    {
      "PartitionKey": "us-east-1:27189d5e-1336-yyyy-yyyy-yyyy-yyyyy-dataset-0306",
      "Data": "eyJpZGVudGl0eVBvb2xJZCI6InVzLWVhc3QtMTphZWM3ZWIzZC0zZjIwLTRjMTItYWU1Yi0zM(略)",
      "SequenceNumber": "49548534993188852874494037325111851874242518733736116226"
    },
    {
      （略）
    }
  ],
  "NextShardIterator": "AAAAAAAAAAGXIBPbDr7yF04Vntv31w+jfSEUazApD2FZ5LD29IuWr+（略）"
}

$ TMP=`aws kinesis get-records --shard-iterator "${KINESIS_SHARD_ITERATOR}"| jq  -r .Records[].Data | base64 -d`

Cognito Streamの登録情報確認

新規レコード追加

echo $TMP | jq '.| select (.syncCount == 1)'
{
  "identityPoolId": "us-east-1:aec7eb3d-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
  "identityId": "us-east-1:27189d5e-yyyy-yyyy-yyyy-yyyyyyyyyyyyy",
  "datasetName": "dataset-0306",
  "operation": "replace",
  "payloadType": "Inline",
  "kinesisSyncRecords": [
    {
      "key": "test-key",
      "value": "test-val",
      "syncCount": 1,
      "lastModifiedDate": 1425620485520,
      "deviceLastModifiedDate": 1425620485520,
      "op": "replace"
    }
  ],
  "kinesisSyncRecordsURL": null,
  "lastModifiedDate": 1425620485520,
  "syncCount": 1
}

レコードの更新

$ echo $TMP | jq '.| select (.syncCount == 2)'
{
  "identityPoolId": "us-east-1:aec7eb3d-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
  "identityId": "us-east-1:27189d5e-yyyy-yyyy-yyyy-yyyyyyyyyyyyy",
  "datasetName": "dataset-0306",
  "operation": "replace",
  "payloadType": "Inline",
  "kinesisSyncRecords": [
    {
      "key": "test-key",
      "value": "update-val",
      "syncCount": 2,
      "lastModifiedDate": 1425620711004,
      "deviceLastModifiedDate": 1425620711004,
      "op": "replace"
    }
  ],
  "kinesisSyncRecordsURL": null,
  "lastModifiedDate": 1425620711004,
  "syncCount": 2
}

レコードの削除

$ echo $TMP | jq '.| select (.syncCount == 3)'
{
  "identityPoolId": "us-east-1:aec7eb3d-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
  "identityId": "us-east-1:27189d5e-yyyy-yyyy-yyyy-yyyyyyyyyyyyy",
  "datasetName": "dataset-0306",
  "operation": "replace",
  "payloadType": "Inline",
  "kinesisSyncRecords": [
    {
      "key": "test-key",
      "value": null,
      "syncCount": 3,
      "lastModifiedDate": 1425620866041,
      "deviceLastModifiedDate": 1425620866041,
      "op": "remove"
    }
  ],
  "kinesisSyncRecordsURL": null,
  "lastModifiedDate": 1425620866041,
  "syncCount": 3
}

JSONレコード操作


$ echo $TMP | jq '.| select (.syncCount == 7)'  
{
  "identityPoolId": "us-east-1:aec7eb3d-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
  "identityId": "us-east-1:27189d5e-yyyy-yyyy-yyyy-yyyyyyyyyyyyy",
  "datasetName": "dataset-0306",
  "operation": "replace",
  "payloadType": "Inline",
  "kinesisSyncRecords": [
    {
      "key": "test2",
      "value": "{\"Menu\":[{\"MenyCode\":1,\"MenuName\":\"まぐろ\",\"MenuPrice\":200},{\"MenyCode\":2,\"MenuName\":\"中トロ\",\"MenuPrice\":300},{\"MenyCode\":3,\"MenuName\":\"サーモン\",\"MenuPrice\":100}]}",
      "syncCount": 7,
      "lastModifiedDate": 1425623513912,
      "deviceLastModifiedDate": 1425623513912,
      "op": "replace"
    }
  ],
  "kinesisSyncRecordsURL": null,
  "lastModifiedDate": 1425623513912,
  "syncCount": 7
}

$ echo $TMP | jq '.| select (.syncCount == 7)' | jq -r .kinesisSyncRecords[].value | jq .
{
  "Menu": [
    {
      "MenyCode": 1,
      "MenuName": "まぐろ",
      "MenuPrice": 200
    },
    {
      "MenyCode": 2,
      "MenuName": "中トロ",
      "MenuPrice": 300
    },
    {
      "MenyCode": 3,
      "MenuName": "サーモン",
      "MenuPrice": 100
    }
  ]
}